package net.bwie.realtime.jtp.dwd.log.function;

import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.jtp.utils.DateTimeUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class AdjLogIsNewProcessFuntion extends KeyedProcessFunction<String,String,String> {

    private transient ValueState<String> firstDataState;

    @Override
    public void open(Configuration parameters) throws Exception {
        firstDataState = getRuntimeContext().getState(new ValueStateDescriptor<String>("firstDataState", String.class));
    }

    @Override
    public void processElement(String s, KeyedProcessFunction<String, String, String>.Context context, Collector<String> collector) throws Exception {
        String value = firstDataState.value();
        JSONObject jsonObject = JSONObject.parseObject(s);
        String str = jsonObject.getJSONObject("common").getString("is_new");
        Long ts = jsonObject.getLong("ts");
        if("1".equals(str)){
            if (null == value){
                firstDataState.update(DateTimeUtil.convertLongToString(ts,DateTimeUtil.DATE_TIME_FORMAT));
            }else {
                if (!value.equals(DateTimeUtil.convertLongToString(ts,DateTimeUtil.DATE_TIME_FORMAT))){
                    jsonObject.getJSONObject("common").put("is_new",0);
                }
            }
        }
        else {
            if (null == value){
                firstDataState.update(DateTimeUtil.convertLongToString(ts - 86400000L,DateTimeUtil.DATE_TIME_FORMAT));
            }
        }
        collector.collect(jsonObject.toJSONString());
    }
}
